package com.game.core.db.service.jedis;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;

import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.SSLParameters;
import javax.net.ssl.SSLSocket;
import javax.net.ssl.SSLSocketFactory;

import org.slf4j.Logger;

import com.game.core.db.common.util.Loggers;
import com.game.core.db.service.jedis.Protocol.Command;
import com.game.core.db.service.jedis.exceptions.JedisConnectionException;
import com.game.core.db.service.jedis.exceptions.JedisDataException;
import com.game.core.db.service.jedis.util.RedisInputStream;
import com.game.core.db.service.jedis.util.RedisOutputStream;
import com.game.core.db.service.jedis.util.SafeEncoder;

public class Connection implements Closeable {

	protected static Logger logger = Loggers.jedisLogger;
	private static final byte[][] EMPTY_ARGS = new byte[0][];

	private String host = Protocol.DEFAULT_HOST;
	private int port = Protocol.DEFAULT_PORT;
	private Socket socket;
	private RedisOutputStream outputStream;
	private RedisInputStream inputStream;
	private int pipelinedCommands = 0;
	private int connectionTimeout = Protocol.DEFAULT_TIMEOUT;
	private int soTimeout = Protocol.DEFAULT_TIMEOUT;
	private boolean broken = false;
	private boolean ssl;
	private SSLSocketFactory sslSocketFactory;
	private SSLParameters sslParameters;
	private HostnameVerifier hostnameVerifier;

	public Connection() {
	}

	public Connection(final String host) {
		this.host = host;
	}

	public Connection(final String host, final int port) {
		this.host = host;
		this.port = port;
	}

	public Connection(final String host, final int port, final boolean ssl) {
		this.host = host;
		this.port = port;
		this.ssl = ssl;
	}

	public Connection(final String host, final int port, final boolean ssl, SSLSocketFactory sslSocketFactory,
			SSLParameters sslParameters, HostnameVerifier hostnameVerifier) {
		this.host = host;
		this.port = port;
		this.ssl = ssl;
		this.sslSocketFactory = sslSocketFactory;
		this.sslParameters = sslParameters;
		this.hostnameVerifier = hostnameVerifier;
	}

	public Socket getSocket() {
		return socket;
	}

	public int getConnectionTimeout() {
		return connectionTimeout;
	}

	public int getSoTimeout() {
		return soTimeout;
	}

	public void setConnectionTimeout(int connectionTimeout) {
		this.connectionTimeout = connectionTimeout;
	}

	public void setSoTimeout(int soTimeout) {
		this.soTimeout = soTimeout;
	}

	public boolean setTimeoutInfinite() {
		try {
			if (!isConnected()) {
				connect();
			}
			socket.setSoTimeout(0);
			return true;
		} catch (SocketException ex) {
			broken = true;
			throw new JedisConnectionException(ex);
		}
	}

	public boolean rollbackTimeout() {
		try {
			socket.setSoTimeout(soTimeout);
			return true;
		} catch (SocketException ex) {
			broken = true;
			throw new JedisConnectionException(ex);
		}
	}

	protected boolean sendCommand(final Command cmd, final String... args) {
		final byte[][] bargs = new byte[args.length][];
		for (int i = 0; i < args.length; i++) {
			bargs[i] = SafeEncoder.encode(args[i]);
		}
		return sendCommand(cmd, bargs);
	}

	protected boolean sendCommand(final Command cmd) {
		return sendCommand(cmd, EMPTY_ARGS);
	}

	protected boolean sendCommand(final Command cmd, final byte[]... args) {
		try {
			if (connect()) {
				Protocol.sendCommand(outputStream, cmd, args);
				pipelinedCommands++;
				//logger.info(cmd.name());
				return true;
			}
		} catch (JedisConnectionException ex) {
			/*
			 * When client send request which formed by invalid protocol, Redis send back
			 * error message before close connection. We try to read it to provide reason of
			 * failure.
			 */
			try {
				String errorMessage = Protocol.readErrorLineIfPossible(inputStream);
				if (errorMessage != null && errorMessage.length() > 0) {
					ex = new JedisConnectionException(errorMessage, ex.getCause());
				}
			} catch (Exception e) {
				/*
				 * Catch any IOException or JedisConnectionException occurred from
				 * InputStream#read and just ignore. This approach is safe because reading error
				 * message is optional and connection will eventually be closed.
				 */
			}
			// Any other exceptions related to connection?
			broken = true;
			// throw ex;
		}
		return false;
	}

	public String getHost() {
		return host;
	}

	public void setHost(final String host) {
		this.host = host;
	}

	public int getPort() {
		return port;
	}

	public void setPort(final int port) {
		this.port = port;
	}

	public boolean connect() {
		boolean ret = true;
		if (!isConnected()) {
			try {
				socket = new Socket();
				socket.setReuseAddress(true);
				socket.setKeepAlive(true); 
				// Will monitor the TCP connection is valid
				socket.setTcpNoDelay(true);
				// Socket buffer Whetherclosed, to ensure timely delivery of data
				socket.setSoLinger(true, 0); 
				// Control calls close () method, the underlying socket is closed immediately

				socket.connect(new InetSocketAddress(host, port), connectionTimeout);
				socket.setSoTimeout(soTimeout);
				
				if(isConnected()) {
					if (ssl) {
						if (null == sslSocketFactory) {
							sslSocketFactory = (SSLSocketFactory) SSLSocketFactory.getDefault();
						}
						socket = (SSLSocket) sslSocketFactory.createSocket(socket, host, port, true);
						if (null != sslParameters) {
							((SSLSocket) socket).setSSLParameters(sslParameters);
						}
						if ((null != hostnameVerifier) && (!hostnameVerifier.verify(host, ((SSLSocket) socket).getSession()))) {
							String message = String.format("The connection to '%s' failed ssl/tls hostname verification.", host);
							throw new JedisConnectionException(message);
						}
					}
					outputStream = new RedisOutputStream(socket.getOutputStream());
					inputStream = new RedisInputStream(socket.getInputStream());
					logger.info(" is success !");
				}else {
					ret = false;
					logger.info(" is filed !");
				}
			} catch (IOException ex) {
				broken = true;
				ret = false;
				throw new JedisConnectionException("Failed connecting to host " + host + ":" + port, ex);
			}
		}
		return ret;
	}

	@Override
	public void close() {
		disconnect();
	}

	@SuppressWarnings("finally")
	public boolean disconnect() {
		if (isConnected()) {
			try {
				outputStream.flush();
				socket.close();
				return true;
			} catch (IOException ex) {
				broken = true;
				throw new JedisConnectionException(ex);
			} finally {
				closeQuietly(socket);
				return true;
			}
		}
		return false;
	}

	public boolean isConnected() {
		return socket != null && socket.isBound() && !socket.isClosed() && socket.isConnected()
				&& !socket.isInputShutdown() && !socket.isOutputShutdown();
	}

	public String getStatusCodeReply() {
		String reply = "";
		flush();
		pipelinedCommands--;
		final byte[] resp = (byte[]) readProtocolWithCheckingBroken();
		if (null != resp) {
			reply = SafeEncoder.encode(resp);
		}
		return reply;
	}

	public String getBulkReply() {
		String reply = "";
		final byte[] result = getBinaryBulkReply();
		if (null != result) {
			reply = SafeEncoder.encode(result);
		}
		return reply;
	}

	public byte[] getBinaryBulkReply() {
		flush();
		pipelinedCommands--;
		return (byte[]) readProtocolWithCheckingBroken();
	}

	public Long getIntegerReply() {
		flush();
		pipelinedCommands--;
		return (Long) readProtocolWithCheckingBroken();
	}

	public List<String> getMultiBulkReply() {
		return BuilderFactory.STRING_LIST.build(getBinaryMultiBulkReply());
	}

	@SuppressWarnings("unchecked")
	public List<byte[]> getBinaryMultiBulkReply() {
		flush();
		pipelinedCommands--;
		return (List<byte[]>) readProtocolWithCheckingBroken();
	}

	public void resetPipelinedCount() {
		pipelinedCommands = 0;
	}

	@SuppressWarnings("unchecked")
	public List<Object> getRawObjectMultiBulkReply() {
		return (List<Object>) readProtocolWithCheckingBroken();
	}

	public List<Object> getObjectMultiBulkReply() {
		flush();
		pipelinedCommands--;
		return getRawObjectMultiBulkReply();
	}

	@SuppressWarnings("unchecked")
	public List<Long> getIntegerMultiBulkReply() {
		flush();
		pipelinedCommands--;
		return (List<Long>) readProtocolWithCheckingBroken();
	}

	public List<Object> getAll() {
		return getAll(0);
	}

	public List<Object> getAll(int except) {
		List<Object> all = new ArrayList<Object>();
		flush();
		while (pipelinedCommands > except) {
			try {
				all.add(readProtocolWithCheckingBroken());
			} catch (JedisDataException e) {
				all.add(e);
			}
			pipelinedCommands--;
		}
		return all;
	}

	public Object getOne() {
		flush();
		pipelinedCommands--;
		return readProtocolWithCheckingBroken();
	}

	public boolean isBroken() {
		return broken;
	}

	protected void flush() {
		try {
			outputStream.flush();
		} catch (IOException ex) {
			broken = true;
			throw new JedisConnectionException(ex);
		}
	}

	protected Object readProtocolWithCheckingBroken() {
		try {
			return Protocol.read(inputStream);
		} catch (JedisConnectionException exc) {
			broken = true;
			throw exc;
		}
	}

	public static void closeQuietly(Socket sock) {
		// It's same thing as Apache Commons - IOUtils.closeQuietly()
		if (sock != null) {
			try {
				sock.close();
			} catch (IOException e) {
				// ignored
				logger.warn("Connection closeQuietly IOException err={}",e.toString());
			}
		}
	}
}
